package com.passion.mq.rocketmq.queue1;

import com.passion.mq.config.RocketMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 类描述：用户操作消息队列配置类
 *
 * @author ZhuYin
 * @since 2024年12月31日
 */
@Slf4j
@Configuration("userOperationRocketMqQueueConfig")
@ConditionalOnBean(RocketMqConfig.class)
public class UserOperationQueueConfig {

    public static String topic;

    @Autowired
    private UserOperationQueueBatchConsumer userOperationQueueBatchConsumer;
    @Autowired
    private UserOperationQueueOrderlyConsumer userOperationQueueOrderlyConsumer;

    @Value("${archives.mq.business[0].queueName}")
    public void setTopic(String topic) {
        UserOperationQueueConfig.topic = topic;
    }

    /**
     * 方法说明：消费用户操作消息队列的消息
     *
     * @return 默认消息消费者
     * @author ZhuYin
     * @since 2024年12月30日
     */
    @Bean
    public DefaultMQPushConsumer consumeUserOperationQueueMsg() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConfig.consumerGroup);
        consumer.setNamesrvAddr(RocketMqConfig.nameServer);
        // 设置批量消费的最大消息数 每次最多消费10条消息
        consumer.setConsumeMessageBatchMaxSize(10);
        // 设置消费线程池大小
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(8);
        // 订阅主题
        consumer.subscribe(topic, "*");
        // 注册消息监听器(批量消费监听器)
        // consumer.registerMessageListener(userOperationQueueBatchConsumeListener);
        // 注册消息监听器(顺序消费监听器)
        consumer.registerMessageListener(userOperationQueueOrderlyConsumer);
        // 启动消费者
        consumer.start();
        return consumer;
    }

}
